package com.shujia.stream

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo2Window1 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("stream")
    //.setMaster("local[2]") //指定两个线程

    val sc = new SparkContext(conf)

    //创建spark streaming上下文对象   指定间隔时间
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    //保存之前每隔batch的计算结果
    ssc.checkpoint("/data/chechpoint1")

    //连接socket创建ds
    val ds = ssc.socketTextStream("192.168.129.101", 7777)

    /**
      * 统计最近15秒单词的数量
      * 每隔5秒统计一次
      *
      */

    val reduceFunc = (x: Int, y: Int) => x + y

    /*   val countDS = ds
         .flatMap(_.split(","))
         .map((_, 1))
         .reduceByKeyAndWindow(
           reduceFunc, //聚合函数
           Durations.seconds(15), //窗口大小
           Durations.seconds(5) //滑动时间
         )*/


    /**
      * 优化之后的窗口操作
      */

    //加函数
    val redueFun = (x: Int, y: Int) => x + y
    //减函数
    val invFun = (i: Int, j: Int) => i - j

    val countDS = ds
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKeyAndWindow(
        redueFun,
        invFun,
        Durations.seconds(15),
        Durations.seconds(5)
      ).filter(_._2 != 0) //过滤没有用的数据

    countDS.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }
}
